package cn.witsky.rtp2ws.udp.impl.netty;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadFactory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.witsky.rtp2ws.Server;
import cn.witsky.rtp2ws.constant.ThreadPoolConstant;
import cn.witsky.rtp2ws.udp.UdpServer;
import cn.witsky.rtp2ws.util.CapcityUtil;
import cn.witsky.rtp2ws.util.DefaultThreadFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollDatagramChannel;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;

public class NettyUdpServer implements UdpServer {
	private static Logger logger = LoggerFactory.getLogger(NettyUdpServer.class);
	private static Map<Integer, Bootstrap> bootstrapMap = null;
	private static int bootstrapCount =2;
	private static ThreadFactory threadFactory = new DefaultThreadFactory(ThreadPoolConstant.udpListener);
	private Channel channnel = null;
	private Integer udpPort;
	private static int threadCount;
	private volatile boolean isStop = false;
	

	
	private static void initBootStrap() {
		int cap=Runtime.getRuntime().availableProcessors();
		threadCount=CapcityUtil.tableSizeFor(cap)/2;
		if (bootstrapMap == null) {
			bootstrapMap = new HashMap<Integer, Bootstrap>();
			threadCount = threadCount / bootstrapCount;
			for (int i = 1; i <= bootstrapCount; i++) {
				Bootstrap b = createBootstrap(i);
				bootstrapMap.put(i, b);
				logger.info("初始化bootstrap-{}......", i);
			}
			logger.info("bootstrap个数:{} threadCount:{} ......", bootstrapMap.size(),threadCount);
		}
	}
	public NettyUdpServer(Integer udpPort) {
		this.udpPort = udpPort;
	}

	public Integer getUdpPort() {
		return udpPort;
	}

	public boolean start() {
		if (isStop) {
			logger.warn("线程已经被销毁了......");
			return false;
		} else {
			return this.bindBootstrap(this.udpPort);
		}
	}

	public void stop() {
		if (!isStop) {
			this.destory();
			isStop = true;
		}
	}

	private static Bootstrap createBootstrap(int index) {
		Bootstrap bootstrap = new Bootstrap();
		if (Epoll.isAvailable()) {
//			bootstrap.group(new EpollEventLoopGroup(Runtime.getRuntime().availableProcessors()*4));
//			bootstrap.group(new MyEpollEventLoopGroup(threadCount, threadFactory,2000000));
			bootstrap.group(new EpollEventLoopGroup(threadCount, threadFactory));
			bootstrap.channel(EpollDatagramChannel.class);
//			bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
//			bootstrap.option(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
			bootstrap.option(EpollChannelOption.MAX_DATAGRAM_PAYLOAD_SIZE, Server.getPayLoadLen()+12);
		} else {
			bootstrap.group(new NioEventLoopGroup(threadCount, threadFactory));
			bootstrap.channel(NioDatagramChannel.class);
		}

		bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
		bootstrap.option(ChannelOption.SO_REUSEADDR, false);
		bootstrap.option(ChannelOption.SO_BROADCAST, true);
		bootstrap.option(ChannelOption.SO_RCVBUF, 1024*8).handler(new ChannelInitializer<Channel>() {
			@Override
			protected void initChannel(Channel channel) throws Exception {
				ChannelPipeline pipeline = channel.pipeline();
				pipeline.addLast(new UdpServerHandler());
			}
		});
		return bootstrap;
	}

	private Bootstrap getBootstrap(Integer udpPort) {
		int index = udpPort % bootstrapCount;
		if (index == 0) {
			index = bootstrapCount;
		}
		return bootstrapMap.get(index);
	}

	private boolean bindBootstrap(Integer udpPort) {
		initBootStrap();
		logger.debug("开始启动接监听udp:{}端口的数据了", this.udpPort);
		ChannelFuture future = null;
		try {
			Bootstrap bootstrap = getBootstrap(udpPort);
			future = bootstrap.bind(new InetSocketAddress(this.getUdpPort())).sync();
			channnel = future.channel();
			channnel.closeFuture().addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture future) throws Exception { // 通过回调只关闭自己监听的channel
					future.channel().close();
				}
			});
			return true;
		} catch (Exception e) {
			logger.error("绑定端口出现异常了", e);
			return false;
		}

	}

	public void destory() {
		logger.info("服务器正在停止监听{}端口", this.udpPort);
		try {
			channnel.close();
			logger.info("服务器停止监听{}端口", this.udpPort);
		} catch (Exception e) {
			logger.error("服务器停止监听" + this.udpPort + "端口出现异常", e);
		}
	}

}
